#ifndef __MY_CHANNEL__
#define __MY_CHANNEL__

#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "../mqcommon/protocol.pb.h"
#include "../mqcommon/helpper.hpp"

#include <iostream>
#include <string>
#include <unordered_map>
#include <mutex>
#include <condition_variable>

#include "./consumer.hpp"

/**
 * 与服务器的需求大致相同，服务器中的信道是为了客户端进行服务，而客户端中的信道是为了用户进行服务
 * 而本项目中客户端的设计，要让用户更加简便的使用，即使这是一个网络服务，用户也无需进行网络相关方面的构建，一切功能本项目都已经实现
 * 用户只需要调用客户端向外暴露出的接口即可
 * 所以客户端的的信道设计不仅是向用户提供使用接口，也是向服务端发送请求的接口
 *
 * 同时客户端的信道也是兼容了消费者的所有功能，而客户端的消费者实际只是一个携带信息的结构，实质上没有任何功能
 *
 * 由于本项目中使用的网络服务是muduo库，muduo库中信息的发送对于发送的执行流来说是异步的，所以假如当客户端发送一个请求之后，muduo库并不会立即发送
 * 这就会导致客户端在向服务器发送创建某些资源（如交换机）的之后，假如其它信道要使用该交换机，而服务器还没有创建好该交换机，这样就会导致服务出现问题
 * 所以当客户端信道在发送出请求之后，必须阻塞等待服务端的对应的响应到来之后才能继续向后执行，所以客户端的信道还需要互斥锁和条件变量
 * 为了便于寻找到对应请求的应答，我们也使用了一个哈希表来记录服务端发送来的响应
 *
 * 同时客户端的信道也需要在析构的时候（也就是信道关闭的时候）检查自己是否是消费者信道，如果是的话就需要删除消费者
 *
 * 而客户端信道向外提供的服务即服务端向外提供的服务
 * 本身的逻辑也很简单，那就是构建对应的请求对象，然后通过连接发送出去即可
 */

namespace NS_Channel
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using BasicCommonResponsePtr = std::shared_ptr<NS_Protocol::BasicCommonResponse>;
    using BasicConsumerResponsePtr = std::shared_ptr<NS_Protocol::BasicConsumerResponse>;
    class Channel
    {
    public:
        using Ptr = std::shared_ptr<Channel>;
        Channel(const muduo::net::TcpConnectionPtr &conn, const ProtobufCodecPtr &codec, const NS_AsyncWorker::AsyncWorker::Ptr& async_worker)
            : _channel_id(NS_Helper::Uuid::uuid()), _conn(conn), _codec(codec), _async_worker(async_worker)
        {
        }

        const std::string &cid()
        {
            return _channel_id;
        }

        bool openChannel()
        {
            std::string rid = NS_Helper::Uuid::uuid();
            NS_Protocol::OpenChannelRequest req;
            req.set_request_id(rid);
            req.set_channel_id(_channel_id);
            _codec->send(_conn, req);
            DLOG("发送创建信道请求");
            BasicCommonResponsePtr resp = waitResponse(rid);
            return resp->ok();
        }

        void closeChannel()
        {
            std::string rid = NS_Helper::Uuid::uuid();
            NS_Protocol::CloseChannelRequest req;
            req.set_request_id(rid);
            req.set_channel_id(_channel_id);
            _codec->send(_conn, req);
            // DLOG("发送删除信道请求");
            waitResponse(rid);
            return;
        }

        // 声明/删除交换机
        bool declareExchange(const std::string &ename,
                             NS_Message::ExchangeType type,
                             bool durable, bool auto_delete,
                             google::protobuf::Map<std::string, std::string> &args)
        {
            // 构建请求结构体
            std::string request_id = NS_Helper::Uuid::uuid();
            NS_Protocol::DeclareExchangeRequest req;

            req.set_request_id(request_id);
            req.set_channel_id(_channel_id);
            req.set_ename(ename);
            req.set_type(type);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);

            // 将请求发送出去
            _codec->send(_conn, req);
            // 阻塞等待请求是否被应答
            DLOG("发送声明交换机请求");
            BasicCommonResponsePtr resp = waitResponse(request_id);
            return resp->ok();
        }
        void deleteExchange(const std::string &ename)
        {
            std::string request_id = NS_Helper::Uuid::uuid();
            NS_Protocol::DeleteExchangeRequest req;

            req.set_request_id(request_id);
            req.set_channel_id(_channel_id);
            req.set_ename(ename);

            // 将请求发送出去
            _codec->send(_conn, req);
            // 阻塞等待请求是否被应答
            DLOG("发送删除交换机请求");
            waitResponse(request_id);
        }
        // 声明/删除消息队列
        // 在声明和删除消息队列的时候需要创建消费者管理的队列单元
        bool declareMsgQueue(const std::string &qname,
                             bool exclusive, bool durable, bool auto_delete,
                             google::protobuf::Map<std::string, std::string> &args)
        {
            // 构建请求结构体
            std::string request_id = NS_Helper::Uuid::uuid();
            NS_Protocol::DeclareMsgQueueRequest req;

            req.set_request_id(request_id);
            req.set_channel_id(_channel_id);
            req.set_qname(qname);
            req.set_exclusive(exclusive);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);

            // 将请求发送出去
            _codec->send(_conn, req);
            // 阻塞等待请求是否被应答
            DLOG("发送声明消息队列请求");
            BasicCommonResponsePtr resp = waitResponse(request_id);
            return resp->ok();
        }
        void deleteMsgQueue(const std::string &qname)
        {
            // 构建请求结构体
            std::string request_id = NS_Helper::Uuid::uuid();
            NS_Protocol::DeleteMsgQueueRequest req;

            req.set_request_id(request_id);
            req.set_channel_id(_channel_id);
            req.set_qname(qname);

            _codec->send(_conn, req);
            // 阻塞等待请求是否被应答
            DLOG("发送删除消息队列请求");
            waitResponse(request_id);
        }
        // 绑定/解绑交换机与消息队列绑定关系
        bool Bind(const std::string ename, const std::string &qname, const std::string &binding_key)
        {
            std::string request_id = NS_Helper::Uuid::uuid();
            NS_Protocol::CreateBindingRequest req;

            req.set_request_id(request_id);
            req.set_channel_id(_channel_id);
            req.set_ename(ename);
            req.set_qname(qname);
            req.set_binding_key(binding_key);

            // 将请求发送出去
            _codec->send(_conn, req);
            // 阻塞等待请求是否被应答
            DLOG("发送建立绑定请求");
            BasicCommonResponsePtr resp = waitResponse(request_id);
            return resp->ok();
        }
        void unBind(const std::string ename, const std::string &qname)
        {
            std::string request_id = NS_Helper::Uuid::uuid();
            NS_Protocol::DeleteBindingRequest req;

            req.set_request_id(request_id);
            req.set_channel_id(_channel_id);
            req.set_ename(ename);
            req.set_qname(qname);

            // 将请求发送出去
            _codec->send(_conn, req);
            // 阻塞等待请求是否被应答
            DLOG("发送解除绑定请求");
            waitResponse(request_id);
        }
        // 消息的发布
        void basicPublic(const std::string &ename, const NS_Message::MessageProperties *mp, const std::string &body)
        {
            std::string request_id = NS_Helper::Uuid::uuid();
            NS_Protocol::BasicPublish req;

            req.set_request_id(request_id);
            req.set_channel_id(_channel_id);
            req.set_ename(ename);
            req.set_body(body);
            // DLOG("是否持久化：%d", mp->mode() == NS_Message::DeliveryMode::DURABLE);
            if (mp != nullptr)
            {
                req.mutable_properties()->set_id(mp->id());
                req.mutable_properties()->set_mode(mp->mode());
                req.mutable_properties()->set_routing_key(mp->routing_key());
            }

            // 将请求发送出去
            _codec->send(_conn, req);
            // 阻塞等待请求是否被应答
            // DLOG("消息发布请求");
            waitResponse(request_id);
        }
        // 消息的确认
        void basicAck(const std::string &msg_id)
        {
            if (_consumer == nullptr)
            {
                DLOG("该信道无消费功能，无法确认消息");
                return;
            }

            std::string request_id = NS_Helper::Uuid::uuid();
            NS_Protocol::BasicAck req;

            req.set_request_id(request_id);
            req.set_channel_id(_channel_id);
            req.set_qname(_consumer->_qname);
            req.set_msg_id(msg_id);

            // 将请求发送出去
            // _async_worker->_thraed_pool.post([this, &req]()
            // {
            //     _codec->send(_conn, req);
            // });
            _codec->send(_conn, req);
            // 阻塞等待请求是否被应答
            // DLOG("消息确认请求");
            waitResponse(request_id);
        }
        // 消息队列的订阅
        void queueSubscription(const std::string &qname, const std::string &consumer_tag, bool auto_ack, NS_Consumer::ConsumerCallback cb)
        {
            if (_consumer != nullptr)
            {
                DLOG("重复订阅消息队列");
                return;
            }
            std::string request_id = NS_Helper::Uuid::uuid();
            NS_Protocol::CreateConsumerRequest req;

            req.set_request_id(request_id);
            req.set_channel_id(_channel_id);
            req.set_qname(qname);
            req.set_consumer_tag(consumer_tag);
            req.set_auto_ack(auto_ack);

            // 将请求发送出去
            _codec->send(_conn, req);
            // 阻塞等待请求是否被应答
            DLOG("订阅消息队列请求");
            BasicCommonResponsePtr resp = waitResponse(request_id);
            if (resp->ok())
            {
                _consumer = std::make_shared<NS_Consumer::Consumer>(consumer_tag, qname, auto_ack, cb);
            }
        }
        // 消息队列的取消订阅
        void queueCancelSubscription()
        {
            if (_consumer == nullptr)
            {
                return;
            }
            std::string request_id = NS_Helper::Uuid::uuid();
            NS_Protocol::DeleteConsumerRequest req;

            req.set_request_id(request_id);
            req.set_channel_id(_channel_id);
            req.set_qname(_consumer->_qname);
            req.set_consumer_tag(_consumer->_tag);

            // 将请求发送出去
            _codec->send(_conn, req);
            // 阻塞等待请求是否被应答
            DLOG("取消订阅消息队列请求");
            waitResponse(request_id);
        }

    public:
        // 服务端发来的应答
        void basicResponse(const BasicCommonResponsePtr &resp)
        {
            std::unique_lock<std::mutex> ul(_mtx);
            // DLOG("应答到来");
            _basic_resp.insert(std::make_pair(resp->request_id(), resp));
            _cv.notify_all();
        }

        void basicConsumerResponse(const BasicConsumerResponsePtr &resp)
        {
            if (_consumer == nullptr)
            {
                DLOG("消息推送目标错误，该信道未订阅消息队列");
                return;
            }
            if(resp == nullptr)
            {
                DLOG("消息应答对象错误");
                return;
            }
            if (_consumer->_tag != resp->consumer_tag())
            {
                DLOG("消息推送目标错误，消费者不一致");
                return;
            }
            // DLOG("消息到来");
            _consumer->_callback(_consumer->_tag, resp->mutable_properties(), resp->body());
        }
        ~Channel() {}

    private:
        BasicCommonResponsePtr waitResponse(const std::string &request_id)
        {
            std::unique_lock<std::mutex> ul(_mtx);
            _cv.wait(ul, [&request_id, this]()
                     { return _basic_resp.find(request_id) != _basic_resp.end(); });
            BasicCommonResponsePtr ret = _basic_resp[request_id];
            _basic_resp.erase(request_id);
            return ret;
        }

    private:
        std::string _channel_id;
        muduo::net::TcpConnectionPtr _conn;
        NS_Consumer::Consumer::Ptr _consumer;
        ProtobufCodecPtr _codec;
        std::mutex _mtx;
        std::condition_variable _cv;
        std::unordered_map<std::string, BasicCommonResponsePtr> _basic_resp;

        NS_AsyncWorker::AsyncWorker::Ptr _async_worker;
    };

    class ChannelManager
    {
    public:
        using Ptr = std::shared_ptr<ChannelManager>;
        ChannelManager() {}
        NS_Channel::Channel::Ptr create(const muduo::net::TcpConnectionPtr &conn,
                                        const ProtobufCodecPtr &codec, const NS_AsyncWorker::AsyncWorker::Ptr& async_worker)
        {
            std::unique_lock<std::mutex> ul(_mutex);
            auto channel = std::make_shared<Channel>(conn, codec, async_worker);
            _channels.insert(std::make_pair(channel->cid(), channel));
            return channel;
        }
        void remove(const std::string &cid)
        {
            std::unique_lock<std::mutex> ul(_mutex);
            _channels.erase(cid);
        }
        NS_Channel::Channel::Ptr get(const std::string &cid)
        {
            std::unique_lock<std::mutex> ul(_mutex);
            auto it = _channels.find(cid);
            if (it == _channels.end())
            {
                return NS_Channel::Channel::Ptr();
            }
            return it->second;
        }

        ~ChannelManager()
        {
        }

    private:
        std::mutex _mutex;
        std::unordered_map<std::string, NS_Channel::Channel::Ptr> _channels;
    };
}

#endif